package com.suzhe.ymall.pay.service.impl;

import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.core.metadata.IPage;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.suzhe.ymall.common.enums.TopicEnum;
import com.suzhe.ymall.common.message.PaySuccessMessage;
import com.suzhe.ymall.common.utils.IdWorker;
import com.suzhe.ymall.pay.enums.PayStatusEnum;
import com.suzhe.ymall.pay.mapper.PayRecordMapper;
import com.suzhe.ymall.pay.pojo.PayRecord;
import com.suzhe.ymall.pay.service.IPayRecordService;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.time.DateUtils;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Service;

import java.util.Date;
import java.util.List;

/**
 * <p>
 * 支付记录表 服务实现类
 * </p>
 *
 * @author suzhe
 * @since 2019-08-13
 */
@Service
@Slf4j
public class PayRecordServiceImpl extends ServiceImpl<PayRecordMapper, PayRecord> implements IPayRecordService {

    private final Integer MAX_SEND_TIME = 8;

    @Autowired
    RocketMQTemplate rocketMQTemplate;

    @Autowired
    IdWorker idWorker;

    @Override
    public void notifyOrder() {

        Date timeLimit = DateUtils.addMinutes(new Date(), -5);
        QueryWrapper<PayRecord> queryWrapper = new QueryWrapper<>();
        queryWrapper.eq("status", PayStatusEnum.TRADE_SUCCESS.value())
                .eq("dead",false) //
                .eq("notified",false)//
                .lt("create_time",timeLimit);
        IPage<PayRecord> page = page(new Page<>(1, 50), queryWrapper);
        List<PayRecord> records = page.getRecords();
        if (!records.isEmpty()){
            records.forEach(item -> {
                item.setSendTimes(item.getSendTimes() + 1);
                if (item.getSendTimes() >= MAX_SEND_TIME){
                    item.setDead(true);
                }
                sendPaySuccess(item);
             });
            updateBatchById(records);
        }
    }

    @Override
    public void sendPaySuccess(PayRecord payRecord) {
        PaySuccessMessage paySuccessMessage = new PaySuccessMessage();
        paySuccessMessage.setMessageId(idWorker.nextId());
        paySuccessMessage.setOrderNo(payRecord.getOrderNo());
        paySuccessMessage.setPayNo(payRecord.getPayNo());
        Message<PaySuccessMessage> message = MessageBuilder.withPayload(paySuccessMessage).build();

        rocketMQTemplate.asyncSend(TopicEnum.PAY_SUCCESS.value(), message, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                //更新支付记录为已通知
                log.info("支付成功消息已发送成功:{}",message);
                updateNotified(payRecord.getPayNo(),true);
            }
            @Override
            public void onException(Throwable e) {
                log.error("发送订单成功消息失败:{}",e);
            }
        });
    }

    @Override
    public void updateNotified(String payNo, boolean notified) {
        PayRecord payRecord = new PayRecord();
        payRecord.setPayNo(payNo);
        payRecord.setNotified(notified);
        updateById(payRecord);
    }

}
